package indicator;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import utils.ImpalaUtil;

import javax.sql.DataSource;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @ClassName IndicatorCommonUVService
 * @Deacription TODO
 * @Author wushumin
 * @Date 2021-06-23 16:51
 * @Version 1.0
 **/
public class IndicatorCommonUVService {
    public static final String INDICATOR_CONFIG_STATE = "IndicatorConfigState";
    public static final String GROUPID = "groupid";
    public static final String PAGEID = "pageid";
    public static final String USER_ID = "user_id";
    public static MapStateDescriptor<String, Map<Long,IndicatorConfig>> INDICATOR_STATE_DESC = new MapStateDescriptor<>(INDICATOR_CONFIG_STATE,
            BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(Long.class,IndicatorConfig.class));

    /**
     * UserEvent处理主流程方法
     * @throws InterruptedException
     */
    public void process() throws InterruptedException {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //step1: indicator stream for braodCast
        BroadcastStream<Map<Long, IndicatorConfig>> indicatorConfigBroadcastStream = environment
                .addSource(new IndicatorConfigSource())
                .broadcast(INDICATOR_STATE_DESC);

        //step2: business datastream - kafka source stream
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "cdh1.lcbint.cn:9092,cdh2.lcbint.cn:9092,cdh3.lcbint.cn:9092");
        properties.setProperty("group.id", "testflink_dev");
        FlinkKafkaConsumer011 kafkaConsumer011 = new FlinkKafkaConsumer011<>("testtopic",new SimpleStringSchema(),properties);
        DataStreamSource kafkaDataStream = environment.addSource(kafkaConsumer011);
        //Map to indicator.UserEvent json
        SingleOutputStreamOperator<JSONObject> eventStream = kafkaDataStream
                .filter(item->item!=null)
                .map(item -> {
            JSONObject userEvent = JSON.parseObject((String) item);
            return userEvent;
        }).returns(JSONObject.class);
        //step4: 过滤并处理数据；
        eventStream.filter(event -> event.getString(GROUPID) != null && event.getString(PAGEID) != null && event.getString(USER_ID) != null
                &&event.getString("event")!=null
                && (event.getString("event").equals("$MPViewScreen")
                    || event.getString("event").equals("$MPShow")
                    || event.getString("event").equals("$MPLaunch")
                ) )
                .keyBy(event -> event.getString(GROUPID) + "_" + event.getString(PAGEID))
                //step4.2: 连接广播 指标配置数据
                .connect(indicatorConfigBroadcastStream)
                //step4.3: 处理数据；
                .process(new CommonUVBroadCastProcessFunction())
                //由于java泛型中将具体类型擦除，所以此处需要返回真实的类型给dataStream
                .returns(String.class)
                .print();
        try {
            environment.execute("common UV indicator cals");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 配置IndicatorConfig数据源
     */
    public static class IndicatorConfigSource extends RichSourceFunction<Map<Long,IndicatorConfig>> {
        private Logger log = LoggerFactory.getLogger(IndicatorConfigSource.class);
        private volatile boolean isRunning = true;
        @Override
        public void run(SourceContext<Map<Long, IndicatorConfig>> ctx) throws Exception {
            DataSource dataSource = ImpalaUtil.getImpalaDataSource();
//            while(isRunning) {
            String indicatorConfigSQL="select * from information_schema.schema_indicator_config_info where metric_code='metric_micro_track_uv' and is_deleted=0 and is_atomic=1 order by indicator_code";
            List<Map<String, Object>> datas = ImpalaUtil.executeQuery(dataSource, indicatorConfigSQL,null);
            if (CollectionUtils.isEmpty(datas)) {
                log.error("======>there is not indicator data");
            }
            Map<Long, IndicatorConfig> dataMap = datas.stream()
                    .filter(item -> item.get("dimension_name") != null && item.get("indicator_table") != null)
                    .map(item -> new IndicatorConfig(Long.valueOf(item.get("id").toString()),
                            item.get("indicator_code").toString(), item.get("dimension_code").toString(),
                            item.get("dimension_name").toString(), item.get("indicator_table").toString(),
                            item.get("indicator_biz_table").toString())
                    ).collect(Collectors.toMap(item -> item.getId(), item -> item));
            log.info("======>query indicator data:{}", JSON.toJSONString(dataMap));
            ctx.collect(dataMap);
//                Thread.sleep(60*1000);
//            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
